package com.github.ghsea.kafka;

import java.util.concurrent.Future;
import javax.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;

/**
 * kafka发送工具类
 */

@Component
@Slf4j
public class KafkaProducer {

    @Resource
    private KafkaProduceContext kafkaProduceContext;

    public Future sendAsync(String topic, Integer partition, String key, String data, Callback callback) {
        Producer<String, String> producer = kafkaProduceContext.getProducer();
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, partition, key, data);
        Future<RecordMetadata> send = producer.send(producerRecord, (recordMetadata, e) -> {
            if (e != null) {
                log.error("kafka topic:{} msg sent failed:{}", recordMetadata.topic(), data, e);
            } else {
                log.info("kafka topic:{} msg sent success:{}", recordMetadata.topic(), data);
            }
            if (callback != null) {
                callback.action(new SendResult<>(producerRecord, recordMetadata), e);
            }
        });
        return send;
    }

    public Future sendAsync(String topic, Integer partition, String key, String data) {
        return sendAsync(topic, partition, key, data, null);
    }

    public Future sendAsync(String topic, String key, String data) {
        return sendAsync(topic, null, key, data, null);
    }

    public interface Callback {
        void action(SendResult<String, String> sendResult, Throwable throwable);
    }

}
